package com.huitone.smspfm.interfacz.mt;

import java.util.ArrayList;
import java.util.Iterator;
import java.util.List;
import java.util.concurrent.ArrayBlockingQueue;

import com.huitone.smspfm.interfacz.logger.ExtLogger;
import com.huitone.smspfm.interfacz.model.MessageSbt;

/**   
* @Title DelRepArrayBlockingQueue.java 
* @Package com.huitone.smspfm.interfacz.delrep 
* @Description TODO(用一句话描述该文件做什么) 
* @author ybliang   
* @date 2017年6月27日 下午4:30:20 
* @version V1.0   
*/
public class SocketServerArrayBlockingQueue {
	
	private static final ExtLogger logger = ExtLogger.create(SocketServerArrayBlockingQueue.class);

	private ArrayBlockingQueue<List<MessageSbt>> queue = null;
	
	private ArrayList<MessageSbt> tmp = null;
	
	private long tmpSize = 2500;
	
	private long lastTime = System.currentTimeMillis();
	
	private long timeInterval = 1000;
	
	private boolean flushOntime = true;
	
	private Thread flushTask = null;

	public SocketServerArrayBlockingQueue(int capacity) {
		queue = new ArrayBlockingQueue<List<MessageSbt>>(capacity);
		this.setFlushOntime(true);
	}

	public synchronized void put(MessageSbt message) {
		
		if (tmp == null) {
			tmp = new ArrayList<MessageSbt>();
		}
		
		if (message != null) {
			tmp.add(message);
		}
		
		long currentTimeMillis = System.currentTimeMillis();
		
		long interval = currentTimeMillis - lastTime;
		
		if (tmp.size() >= tmpSize || interval >= timeInterval) {
			
			if (!tmp.isEmpty()) {
				ArrayList<MessageSbt> messages = tmp;
				try {
					queue.put(messages);
					tmp = null;
					lastTime = currentTimeMillis;
				} catch (InterruptedException e) {
					logger.debug(e.getMessage(),e);
				}
			} else {
				lastTime = currentTimeMillis;
			}
		}
	}
	
	public synchronized void flush() {
		
		long currentTimeMillis = System.currentTimeMillis();
		
		if (tmp != null && !tmp.isEmpty()) {
			ArrayList<MessageSbt> messages = new ArrayList<MessageSbt>();
			messages.addAll(tmp);
			try {
				queue.put(messages);
				tmp.clear();
				lastTime = currentTimeMillis;
			} catch (InterruptedException e) {
				logger.debug(e.getMessage(),e);
			}
		} else {
			lastTime = currentTimeMillis;
		}
	}
	
	public List<MessageSbt> take() {
		
		List<MessageSbt> result = null;
		
		try {
			result = queue.take();
		} catch (InterruptedException e) {
			logger.debug(e.getClass().getName());
		}
		
		return result;
	}
	
	public List<MessageSbt> poll() {
		return queue.poll();
	} 
	
	public int getSize() {
		return queue.size();
	}
	
	public long getElementCount() {
		
		int count = 0;
		
		Iterator<List<MessageSbt>> iterator = queue.iterator();
		while (iterator.hasNext()) {
			List<MessageSbt> list = iterator.next();
			count += list.size();
		}
		
		return count;
	}

	public boolean isFlushOntime() {
		return flushOntime;
	}

	public void setFlushOntime(boolean flushOntime) {
		this.flushOntime = flushOntime;
		if (isFlushOntime()) {
			if (flushTask == null || !flushTask.isAlive()) {
				flushTask = new Thread(new Runnable() {
					
					@Override
					public void run() {
						
						while (!Thread.currentThread().isInterrupted() && isFlushOntime()) {
							
							SocketServerArrayBlockingQueue.this.put(null);
							
							try {
								Thread.sleep(100);
							} catch (InterruptedException e) {
								break;
							}
						}
						
					}
				});
				flushTask.start();
			}
		} else {
			if (flushTask != null) {
				flushTask.interrupt();
				flushTask = null;
			}
		}
	}
	
	public void destroy() {
		this.flush();
		this.setFlushOntime(false);
	}
}
